package com.platform.apex.input;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.LinkedBlockingQueue;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.InputOperator;
import com.datatorrent.common.util.BaseOperator;

public class MySqlInputOperator extends BaseOperator implements InputOperator {

	protected transient MyJdbcStore store;
	protected transient LinkedBlockingQueue emitQueue;
	public final transient DefaultOutputPort<Object> outputPort = new DefaultOutputPort<Object>();
	private String databaseUrl;
	private String userName;
	private String password;
	private int currentNumber;
	private String databaseDriver;
	
	

	public MySqlInputOperator() {
	}

	@Override
	public void setup(OperatorContext context) {
		// TODO Auto-generated method stub
		super.setup(context);
		store = new MyJdbcStore(databaseUrl, userName, password,databaseDriver);
		emitQueue = new LinkedBlockingQueue<>(10000);
	}

	@Override
	public void beginWindow(long windowId) {
		// TODO Auto-generated method stub
		super.beginWindow(windowId);
		String sql = "select a.tran_id,a.work_id,a.flow_id,a.node_id,a.step_id,a.detail_id,a.belong_work,a.tran_title,a.timely_time"
				+ ",a.timely_status,a.limit_time,a.attact_qty,a.attact_content,a.create_dept_id,a.create_dept_name,a.create_opt_id"
				+ ",a.create_opt_name,a.create_opt_time,a.deal_start_time,a.deal_part_id,a.deal_part_name,a.deal_opt_id,a.deal_opt_name"
				+ ",a.deal_opt_time,a.deal_time,a.over_status,d.deal_status,a.work_status,a.deal_view,a.supervise_qty,a.urge_qty,a.warn_qty,a.opt_order"
				+ ",a.next_dept,a.next_person,a.pre_tran_id,a.dispatch_opt_id,a.dispatch_opt_name,a.data_resource,a.data_type,b.warn_id,b.warn_title,b.warn_content"
				+ ",b.warn_type,b.opt_id,b.opt_name,b.receive_opt_id,b.receive_opt_name,b.opt_time,b.warn_sort,b.is_close,b.close_opt_id,b.close_opt_name,b.close_opinion"
				+ ",b.close_opt_time,c.supervise_id,c.supervise_type,c.supervise_title,c.supervise_opinion,c.is_warn,c.back_opt_id,c.back_opt_name,c.back_status"
				+ ",c.back_opinion,c.back_opt_time,c.is_valid,c.deal_result,c.adjust_flow,c.supervise_level,c.remark,d.urge_id,d.urge_title,d.urge_opinion,d.wish_opt_time"
				+ ",d.receive_status,d.receive_opinion,d.receive_opt_time,d.deal_opinion from dh_tranlist a left JOIN dh_warning b ON a.tran_id = b.tran_id"
				+ " LEFT JOIN dh_supervise c on a.tran_id = c.tran_id"
				+ " LEFT JOIN dh_urge d on a.tran_id = d.tran_id ORDER BY a.tran_id asc LIMIT %s,100";
		
		 sql = String.format(sql, String.valueOf(currentNumber*100));
		queryAndInsert(sql);
	}

	@Override
	public void endWindow() {
		// TODO Auto-generated method stub
		super.endWindow();
		this.currentNumber++;
		
	}

	@Override
	public void emitTuples() {
		// TODO Auto-generated method stub
		if(emitQueue.size()>0){
			try {
				outputPort.emit(emitQueue.take());
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

	}

	public MyJdbcStore getStore() {
		return store;
	}

	public void setStore(MyJdbcStore store) {
		this.store = store;
	}

	public void queryAndInsert(String querySql) {

		

		Connection connSelect = null;
		PreparedStatement ps = null;
		ResultSet rs = null;

		Map<String, String> mapToes = null;
		
			try {
				
				store.connect();
				connSelect = store.getConnection();

				ps = connSelect.prepareStatement(querySql);

				rs = ps.executeQuery();

				ResultSetMetaData rsmd = rs.getMetaData();
				// 可以得到有多少列
				int columnNum = rsmd.getColumnCount();
				while (rs.next()) {
					Object[] objects = new Object[columnNum];
					mapToes = new HashMap<String, String>();
					for (int i = 0; i < objects.length; i++) {
						objects[i] = rs.getObject(i + 1);
						String columnName = rsmd.getColumnName(i + 1);

						mapToes.put(columnName, String.valueOf(objects[i]));
					}
					Map<String, String> sortMapByKey = sortMapByKey(mapToes);
					emitQueue.put(sortMapByKey);
				}
				ps.close();
				connSelect.close();
			} catch (Exception e) {
				// TODO: handle exception
				e.printStackTrace();
			}
	}

	


	public String getDatabaseUrl() {
		return databaseUrl;
	}

	public void setDatabaseUrl(String databaseUrl) {
		this.databaseUrl = databaseUrl;
	}

	public String getDatabaseDriver() {
		return databaseDriver;
	}

	public void setDatabaseDriver(String databaseDriver) {
		this.databaseDriver = databaseDriver;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	

	public static Map<String, String> sortMapByKey(Map<String, String> map) {
		if (map == null || map.isEmpty()) {
			return null;
		}

		Map<String, String> sortMap = new TreeMap<String, String>(new MapKeyComparator());

		sortMap.putAll(map);

		return sortMap;
	}
	
	

}
